package com.enjoyor.soa.traffic.server.efoms.dubbo.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.enjoyor.soa.traffic.server.efoms.EfomsServerApplication;
import com.enjoyor.soa.traffic.server.efoms.dubbo.invoke.DubboServerPushInvoke;
import com.enjoyor.soa.traffic.util.pojo.ResultPojo;
import org.apache.log4j.Logger;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
@Component
public class ServerpushBlock {

    private static final Logger logger = Logger.getLogger(ServerpushBlock.class);

    private static final BlockingQueue<Map<String, Object>> msgQueue = new LinkedBlockingQueue<>();

    private static ServerpushBlock instance;
    private DubboServerPushInvoke getServerPushService() {
    	DubboServerPushInvoke service = EfomsServerApplication.context.getBean(DubboServerPushInvoke.class);
		return service;
	}
    
    @Scheduled(fixedDelay = 5000)
    public ServerpushBlock getInstance() {
        if (instance == null) {
            synchronized (ServerpushBlock.class) {
                if (instance == null) {
                    instance = new ServerpushBlock();
                    Thread thread = new Thread(new PushMessageThread());
                    logger.info("ServerpushBlock Listener is run -------->>>");
                    thread.setName("T-->ServerPush");
                    thread.start();
                }
            }
        }
        return instance;
    }

    class PushMessageThread implements Runnable {
        @Override
        public void run() {
            while (true) {
                Map<String, Object> messageInfoDo = getInstance().getMessageQueue().poll();
                if (messageInfoDo != null) {
                    getInstance().notifyListener(messageInfoDo);
                }
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }

    }

    /**
     * 处理通道内信息
     *
     * @param msg
     */
    public void notifyListener(Map<String, Object> msg) {
        try {
        	JSONObject json = JSONObject.parseObject(JSON.toJSONString(msg));
            ResultPojo resultPojo = instance.getServerPushService().pushMessage((String) msg.get("OPRATION_NAME"), json);
            
            //System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())
            //        +"send message: {"+msg.get("OPRATION_NAME")+","+ json+"}");
            logger.info(resultPojo.convertToJson().toJSONString());
        } catch (Exception e) {
            this.msgQueue.remove(msg);
            logger.error("send message is error :{}",e);
        }
    }

    public BlockingQueue<Map<String, Object>> getMessageQueue() {
        return this.msgQueue;
    }


    public static void addMessage(Map<String, Object> message) {
        msgQueue.add(message);
    }
}
